package com.company.beauty.ids.handler;

import cn.hutool.extra.spring.SpringUtil;
import com.company.beauty.ids.util.IpUtils;
import com.company.beauty.ids.util.PortUtils;
import com.company.nuwa.common.key.ApplicationNamedKeyGenerator;
import com.company.nuwa.common.utils.Preconditions;
import com.company.nuwa.redis.support.lock.RedisLocker;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * <p>snowflake redis handler</p >
 * ids-starter 提供 redis实现，若个别服务，需要使用zookeeper等方案实现，可以新增类实现 {@link SnowflakeHandler}
 * 只需要确保 {@code weight} 权重大于当前redis实现即可
 *
 * @author youyoukele
 * @version 1.0
 * @date 2024/1/4 10:58
 */
@Slf4j
public class SnowflakeRedisHandler implements SnowflakeHandler {

    private static final int LOAD_COUNT = 3;

    private StringRedisTemplate stringRedisTemplate;
    private String ip;
    private String port;

    private int workerId = -1;
    private int centerId = -1;

    /**
     * nodeFlag / 32 为 centerId
     * nodeFlag % 32 为 workId
     */
    private int nodeFlag = -1;

    public SnowflakeRedisHandler() {
        this.init();
    }

    @Override
    public int weight() {
        return SnowflakeHandler.super.weight();
    }

    @Override
    public int getWorkerId() {
        return workerId;
    }

    @Override
    public int getCenterId() {
        return centerId;
    }

    /**
     * init
     */
    private void init() {
        this.stringRedisTemplate = SpringUtil.getBean(StringRedisTemplate.class);
        this.ip = IpUtils.getIp();
        this.port = String.valueOf(PortUtils.getPort());
        // 分配节点编码，极端情况有可能多个节点并发启动，导致分配异常
        // 此处尝试多次，提高服务启动的容错性
        boolean result = false;
        for (int i = 0; i < LOAD_COUNT; i++) {
            if (this.assign()) {
                result = true;
                break;
            }
        }
        Preconditions.checkArgument(result, -1, "ids starter 多个服务并发启动导致节点编码一致，需要重新启动!!!");
        // 心跳续约,每30秒一次
        this.scheduledRenewal(workerId);
    }

    /**
     * 节点实例，每个节点获取的workerId,和centerId 需要保持唯一
     *
     * @return
     */
    private String getNode() {
        return ip + ":" + port;
    }

    /**
     * 分配 worker id, center id
     *
     * @return
     */
    private boolean assign() {
        log.info("当前机器ip:{}, port:{}", ip, port);

        String node = this.getNode();

        // 终端ID一共32个, 从0~31,
        // 集群中心一共32, 从0~31
        // 待使用的nodeFlag列表
        List<Integer> waiteUsedNodeFlags = Lists.newArrayList();
        for (int remoteNodeFlag = 0; remoteNodeFlag < (32 * 32) - 1; remoteNodeFlag++) {
            String remoteNode = stringRedisTemplate.opsForValue().get(CacheKeys.genNodeFlagMappingNodeKey(remoteNodeFlag));
            // 如果当前node已占用某个nodeFlag，则直接沿用。
            if (StringUtils.equals(remoteNode, node)) {
                nodeFlag = remoteNodeFlag;
                log.info("如果当前node已占用某个nodeFlag，则直接沿用。 nodeFlag:{}, node:{}", nodeFlag, node);
                break;
            } else if (StringUtils.isBlank(remoteNode)) {
                // 如果发现某个nodeFlag没有被占用，则先记录。
                waiteUsedNodeFlags.add(remoteNodeFlag);
            }
        }
        // 如果当前node没有占用某个nodeFlag, 则从待使用的列表中分配
        if (nodeFlag == -1 && !waiteUsedNodeFlags.isEmpty()) {
            for (Integer waiteUsedNodeFlag : waiteUsedNodeFlags) {
                // 这里是为了提高容错性，避免因为并发启动服务，导致多个节点分配同一nodeFlag，但是在高并发情况下无法保证百分百准确
                // 理论上来讲，高并发启动服务的可能性不大，暂不考虑使用分布式锁
                // todo youyoukele 后续若有需要可以考虑加分布式锁
                if (Boolean.FALSE.equals(stringRedisTemplate.hasKey(CacheKeys.genNodeFlagMappingNodeKey(waiteUsedNodeFlag)))) {
                    nodeFlag = waiteUsedNodeFlag;
                    log.info("如果当前node没有占用某个nodeFlag, 则从待使用的列表中分配。 nodeFlag:{}, node:{}", nodeFlag, node);
                    break;
                }
            }
            waiteUsedNodeFlags.remove(nodeFlag);
            log.info("待使用的nodeFlag数:{}", waiteUsedNodeFlags.size());
        }

        // nodeFlag 通过redis获取异常，此处需要排查
        if (nodeFlag == -1) {
            log.error("nodeFlag 通过redis获取异常，此处需要排查。");
            return Boolean.FALSE;
        } else {
            centerId = nodeFlag / 32;
            workerId = nodeFlag % 32;
        }

        // 当前机器节点占用该坑位1小时
        String key = CacheKeys.genNodeFlagMappingNodeKey(nodeFlag);
        RedisLocker redisLocker = SpringUtil.getBean(RedisLocker.class);
        boolean result = redisLocker.tryLock(CacheKeys.genNodeFlagMappingLockerKey(), () -> {
            String remoteNode = stringRedisTemplate.opsForValue().get(key);
            if (Objects.isNull(remoteNode) || StringUtils.equals(node, remoteNode)) {
                // 远程节点未占用，或者远程节点即当前节点
                stringRedisTemplate.opsForValue().set(key, node, 1, TimeUnit.HOURS);
                log.info("nodeFlag 初始化 {} {} workerId:{} centerId:{}", key, node, workerId, centerId);
                return Boolean.TRUE;
            } else {
                return Boolean.FALSE;
            }
        }, () -> {
            // 之所以在这里兜底，是为了提高服务启动的并发性，理论上到达这一步的可能性不大
            log.error("ids starter 多个服务并发启动导致节点编码一致，请重试");
            return Boolean.FALSE;
        });
        return result;
    }

    /**
     * 心跳续约,每30秒一次
     *
     * @param nodeFlag
     */
    private void scheduledRenewal(long nodeFlag) {
        String key = CacheKeys.genNodeFlagMappingNodeKey(nodeFlag);

        String node = this.getNode();

        // 心跳续约,每30秒一次
        Executors.newSingleThreadScheduledExecutor(r -> {
            Thread thread = new Thread(r, "schedule-renewal");
            thread.setDaemon(true);
            return thread;
        }).scheduleWithFixedDelay(() -> {
            stringRedisTemplate.opsForValue().set(key, node, 1, TimeUnit.HOURS);
            log.info("nodeFlag 续约 {} {} workerId:{} centerId:{}", key, node, workerId, centerId);
        }, 0, 30, TimeUnit.SECONDS);
    }

    /**
     * <p></p >
     *
     * @author youyoukele
     * @version 1.0
     * @date 2024/1/4 10:58
     */
    private static class CacheKeys {
        /**
         * 节点映射key
         *
         * @param nodeFlag
         * @return
         */
        private static String genNodeFlagMappingNodeKey(long nodeFlag) {
            return ApplicationNamedKeyGenerator.genKey(true, "node-flag-mapping-node", String.valueOf(nodeFlag));
        }

        /**
         * 生成节点时的锁key
         *
         * @return
         */
        private static String genNodeFlagMappingLockerKey() {
            return ApplicationNamedKeyGenerator.genKey(true, "node-flag-mapping-locker");
        }
    }

}
